package org.yuanzheng.tableAndSql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.types.Row
import org.yuanzheng.source.{CustomSource, StationLog}

/*批计算*/
object TestTableApi {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //读取数据
    val stream: DataStream[StationLog] = streamEnv.addSource(new CustomSource)
    val table: Table = tableEnv.fromDataStream(stream)

    //过滤
    //val res: Table = table.filter('callType === "success")
    val res: Table = table.where('callType === "success")
    val ds: DataStream[Row] = tableEnv.toAppendStream[Row](res.select('sid, 'callType as 'type, 'callTime, 'callOut))
    ds.print("where")
    //分组聚合
    val result: Table = table.groupBy('sid).select('sid, 'sid.count as 'sidCount)
    val value = tableEnv.toRetractStream[Row](result)
    value.filter(_._1 == true) print ("groupBy")

    streamEnv.execute()
  }
}
